package com.caius.xiaohashu.note.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.note.biz.domain.dataobject.NoteLikeDO;
import com.caius.xiaohashu.note.biz.domain.mapper.NoteLikeDOMapper;
import com.caius.xiaohashu.note.biz.model.dto.LikeUnlikeNoteReqDTO;
import com.google.common.collect.Lists;
import constant.MQConstants;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author Caius
 * @description
 * @since Created in 2025-05-27
 */
@Component
@Slf4j
public class LikeUnlikeNoteConsumer {

    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    @Resource
    private NoteLikeDOMapper noteLikeDOMapper;

    private DefaultMQPushConsumer consumer;

    // 初始化 5000 个令牌桶
    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Bean(name = "LikeUnlikeNoteConsumer")
    public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {
        String group = "xiaohashu_group_" + MQConstants.TOPIC_LIKE_OR_UNLIKE;

        consumer = new DefaultMQPushConsumer(group);

        consumer.setNamesrvAddr(namesrvAddr);

        consumer.subscribe(MQConstants.TOPIC_COUNT_NOTE_LIKE, "*");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.setMessageModel(MessageModel.CLUSTERING);

        consumer.setMaxReconsumeTimes(3);

        consumer.setConsumeMessageBatchMaxSize(30);

        consumer.setPullInterval(1000);

        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            log.info("==> 【笔记点赞、取消点赞】本批次消息大小: {}", msgs.size());
            try {
                rateLimiter.acquire();

                List<LikeUnlikeNoteReqDTO> likeUnlikeNoteReqDTOS = Lists.newArrayList();
                msgs.forEach(msg -> {
                    String msgJson = new String(msg.getBody());
                    log.info("==> Consumer - Received message: {}", msgJson);
                    likeUnlikeNoteReqDTOS.add(JsonUtils.parseObject(msgJson, LikeUnlikeNoteReqDTO.class));
                });

                // 幂等性: 通过联合唯一索引保证

                // 内存级操作
                // ID 分组
                Map<Long, List<LikeUnlikeNoteReqDTO>> groupMap = likeUnlikeNoteReqDTOS.stream()
                        .collect(Collectors.groupingBy(LikeUnlikeNoteReqDTO::getUserId));

                // 通过操作再次分组
                List<LikeUnlikeNoteReqDTO> finalOperations = groupMap.values().stream()
                        .flatMap(userOperations -> {
                            Map<Long, List<LikeUnlikeNoteReqDTO>> noteGroupMap = userOperations.stream()
                                    .collect(Collectors.groupingBy(LikeUnlikeNoteReqDTO::getNoteId));

                            return noteGroupMap.entrySet().stream()
                                    .filter(entry -> {
                                        List<LikeUnlikeNoteReqDTO> operations = entry.getValue();
                                        int size = operations.size();
                                        if (size % 2 == 0) {
                                            // 偶数次操作取消
                                            return false;
                                        } else {
                                            return true;
                                        }
                                    })
                                    .map(entry -> {
                                        List<LikeUnlikeNoteReqDTO> ops = entry.getValue();
                                        // 取最后一次操作
                                        return ops.get(ops.size() - 1);
                                    });
                        }).toList();
                // 2. 批量写入数据库
                if (CollUtil.isNotEmpty(finalOperations)) {
                    List<NoteLikeDO> noteLikeDOS = finalOperations.stream()
                            .map(finalOperation -> NoteLikeDO.builder()
                                    .userId(finalOperation.getUserId())
                                    .noteId(finalOperation.getNoteId())
                                    .createTime(finalOperation.getCreateTime())
                                    .status(finalOperation.getType())
                                    .build())
                            .toList();

                    noteLikeDOMapper.batchInsertOrUpdate(noteLikeDOS);
                }

                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                log.error("", e);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        });
        consumer.start();
        return consumer;
    }

    @PreDestroy
    public void destroy() {
        if (Objects.nonNull(consumer)) {
            try {
                consumer.shutdown();
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }
}